package com.uzai.console.constant.ES;

import cn.hutool.core.util.NumberUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.uzai.common.exception.BusinessException;
import com.uzai.common.result.ResponseCode;
import com.uzai.console.config.ElasticsearchBaseConfig;
import com.uzai.console.config.ElasticsearchGroupMsgConfig;
import com.uzai.console.constant.Const;
import com.uzai.console.service.alarm.impl.AlarmContacterServiceImpl;
import org.apache.commons.lang3.ArrayUtils;
import org.apache.commons.lang3.ObjectUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.lucene.search.TotalHits;
import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.bulk.BulkResponse;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.search.ClearScrollRequest;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.action.search.SearchScrollRequest;
import org.elasticsearch.action.support.WriteRequest;
import org.elasticsearch.action.support.master.AcknowledgedResponse;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.indices.CreateIndexRequest;
import org.elasticsearch.client.indices.GetIndexRequest;
import org.elasticsearch.common.Strings;
import org.elasticsearch.common.text.Text;
import org.elasticsearch.common.unit.DistanceUnit;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.core.TimeValue;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.GeoDistanceQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.reindex.BulkByScrollResponse;
import org.elasticsearch.index.reindex.DeleteByQueryRequest;
import org.elasticsearch.search.Scroll;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.aggregations.Aggregation;
import org.elasticsearch.search.aggregations.AggregationBuilder;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.Aggregations;
import org.elasticsearch.search.aggregations.bucket.histogram.Histogram;
import org.elasticsearch.search.aggregations.metrics.Cardinality;
import org.elasticsearch.search.aggregations.metrics.ParsedCardinality;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.fetch.subphase.FetchSourceContext;
import org.elasticsearch.search.fetch.subphase.highlight.HighlightBuilder;
import org.elasticsearch.search.fetch.subphase.highlight.HighlightField;
import org.elasticsearch.search.sort.SortOrder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.io.IOException;
import java.util.*;

/**
 * @author liuqi
 * @date 2021年11月20日 14:51
 */
@Component
public class ElasticsearchUtilGroupMsg {
     Logger log = LoggerFactory.getLogger(AlarmContacterServiceImpl.class);

    //最好不要自定义id 会影响插入速度。

    @Autowired
    @Qualifier(ElasticsearchGroupMsgConfig.NAME)
    private  RestHighLevelClient restHighLevelClientGroupMsg;

    /**
     * 创建索引
     * @param index
     * @return
     */
    public  boolean createIndex(String index){
        if(isIndexExist(index)){
            log.error("Index is  exits!");
            return false;
        }
        //1.创建索引请求
        CreateIndexRequest request = new CreateIndexRequest(index);
        //2.执行客户端请求
        try {
            org.elasticsearch.client.indices.CreateIndexResponse response = restHighLevelClientGroupMsg.indices()
                    .create(request, RequestOptions.DEFAULT);
            return response.isAcknowledged();
        } catch (IOException e) {
        }
        return false;
    }

    /**
     * 判断索引是否存在
     * @param index
     * @return
     */
    public   boolean isIndexExist(String index) {
        GetIndexRequest request = new GetIndexRequest(index);
        boolean exists = false;
        try {
            exists = restHighLevelClientGroupMsg.indices().exists(request, RequestOptions.DEFAULT);
        } catch (IOException e) {
        }
        return exists;
    }

    /**
     * 删除索引
     * @param index
     * @return
     */
    public  boolean deleteIndex(String index){
        if(!isIndexExist(index)) {
            log.error("Index is not exits!");
            return false;
        }
        DeleteIndexRequest request = new DeleteIndexRequest(index);
        try {
            AcknowledgedResponse delete = restHighLevelClientGroupMsg.indices()
                    .delete(request, RequestOptions.DEFAULT);
            return delete.isAcknowledged();
        } catch (IOException e) {
        }
        return false;
    }

    /**
     * 数据添加，自定义id
     * @param object 要增加的数据
     * @param index      索引，类似数据库
     * @param id         数据ID,为null时es随机生成
     * @return
     */
    public  String addData(Object object, String index, String id) {
        //创建请求
        IndexRequest request = new IndexRequest(index);
        //规则 put /test_index/_doc/1
        request.id(id);
        request.timeout(TimeValue.timeValueSeconds(1));
        //将数据放入请求 json
        IndexRequest source = request.source(JSON.toJSONString(object), XContentType.JSON);
        //客户端发送请求
        IndexResponse response = null;
        try {
            response = restHighLevelClientGroupMsg.index(request, RequestOptions.DEFAULT);
        } catch (IOException e) {
        }
        return response.getId();
    }

    /**
     * 批量数据添加，自定义id
     * @param list 要增加的数据列表
     * @param index      索引，类似数据库
     * @return
     */
    public boolean addDataBatch(List<Map<String, Object>> list , String index) {
        if (ObjectUtils.isEmpty(list)){
            return false;
        }

        int size = list.size();
        BulkRequest request = new BulkRequest();
        for (int i = 0; i < size; i++) {
            Map<String, Object> map = list.get(i);
            request.add(new IndexRequest(index, "_doc", String.valueOf(map.get("id"))).source(map, XContentType.JSON));
        }
        try {
            request.setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE);
            BulkResponse bulk = restHighLevelClientGroupMsg.bulk(request, RequestOptions.DEFAULT);
            return true;
        } catch (IOException e) {
            return false;
        }
    }

    /**
     * 数据添加 随机id
     * @param object 要增加的数据
     * @param index      索引，类似数据库
     * @return
     */
    public String addData(Object object, String index){
        return addData(object, index, UUID.randomUUID().toString().replaceAll("-", "").toUpperCase());
    }

    /**
     * 通过ID删除数据
     * @param index 索引，类似数据库
     * @param id    数据ID
     * @return
     */
    public   void deleteDataById(String index, String id) {
        DeleteRequest request = new DeleteRequest(index).id(id);
        try {
            restHighLevelClientGroupMsg.delete(request, RequestOptions.DEFAULT);
        } catch (IOException e) {
            log.error(e.getMessage(),e);
        }
    }

    /**
     * 删除符合单条件所有数据
     * @param index 索引，类似数据库
     * @param deleteByQueryRequest    数据ID
     * @return
     */
    public void deleteDataByQuerySingle(String index, DeleteByQueryRequest deleteByQueryRequest) {
        deleteByQueryRequest.indices(index);
        try {
            //分词式删除
            BulkByScrollResponse resp =  restHighLevelClientGroupMsg.deleteByQuery(deleteByQueryRequest, RequestOptions.DEFAULT);
        } catch (IOException e) {
            log.error(e.getMessage(),e);
        }
    }

    /**
     * 删除符合多条件所有数据
     * @param index 索引，类似数据库
     * @param deleteByQueryRequest    数据ID
     * @return
     */
    public void deleteDataByQueryMulti (String index, DeleteByQueryRequest deleteByQueryRequest) {
        deleteByQueryRequest.indices(index);
        try {
            //分词式删除
            BulkByScrollResponse resp =  restHighLevelClientGroupMsg.deleteByQuery(deleteByQueryRequest, RequestOptions.DEFAULT);
        } catch (IOException e) {
            log.error(e.getMessage(),e);
        }
    }



    /**
     * 通过ID 更新数据
     * @param object     要更新数据
     * @param index      索引，类似数据库
     * @param id         数据ID
     * @return
     */
    public  void updateDataById(Object object, String index, String id) {
        UpdateRequest update = new UpdateRequest(index, id);
        update.timeout("1s");
        update.doc(JSON.toJSONString(object), XContentType.JSON);
        //可以修改null
        //update.doc(JSON.toJSONString(object, SerializerFeature.WriteMapNullValue), XContentType.JSON);
        try {
            restHighLevelClientGroupMsg.update(update, RequestOptions.DEFAULT);
        } catch (IOException e) {
            log.error(e.getMessage(),e);
        }
    }

    /**
     * 通过ID 更新数据,可以将数据修改为null
     * @param object     要更新数据
     * @param index      索引，类似数据库
     * @param id         数据ID
     * @return
     */
    public  void updateDataNullById(Object object, String index, String id) {
        UpdateRequest update = new UpdateRequest(index, id);
        update.timeout("1s");
        //update.doc(JSON.toJSONString(object), XContentType.JSON);
        //可以修改null
        update.doc(JSON.toJSONString(object, SerializerFeature.WriteMapNullValue), XContentType.JSON);
        try {
            restHighLevelClientGroupMsg.update(update, RequestOptions.DEFAULT);
        } catch (IOException e) {
            log.error(e.getMessage(),e);
        }
    }


    /**
     * 通过ID获取数据
     * @param index  索引，类似数据库
     * @param id     数据ID
     * @param fields 需要显示的字段，逗号分隔（缺省为全部字段）
     * @return
     */
    public  Map<String,Object> searchDataById(String index, String id, String fields){
        Map<String,Object> dataMap = new HashMap<>();
        try {
            GetRequest request = new GetRequest(index, id);
            if (StringUtils.isNotEmpty(fields)){
                //只查询特定字段。如果需要查询所有字段则不设置该项。
                request.fetchSourceContext(new FetchSourceContext(true,fields.split(","), Strings.EMPTY_ARRAY));
            }

            //打印查询ES条件
            //log.info(request.toString());

            GetResponse response = restHighLevelClientGroupMsg.get(request, RequestOptions.DEFAULT);
            dataMap =  response.getSource();
        }catch (IOException e){
            log.error("---ES searchDataById error; index={}, id={}, error={}",index, id , e.getMessage());
        }
        return dataMap;
    }

    /**
     * 通过ID判断文档是否存在
     * @param index  索引，类似数据库
     * @param id     数据ID
     * @return
     */
    public  boolean existsById(String index,String id) throws IOException {
        GetRequest request = new GetRequest(index, id);
        //不获取返回的_source的上下文
        request.fetchSourceContext(new FetchSourceContext(false));
        request.storedFields("_none_");
        return restHighLevelClientGroupMsg.exists(request, RequestOptions.DEFAULT);
    }

    /**
     * 批量插入false成功
     * @param index  索引，类似数据库
     * @param objects     数据
     * @return
     */
    public   boolean bulkPost(String index, List<?> objects) {
        BulkRequest bulkRequest = new BulkRequest();
        BulkResponse response=null;
        //最大数量不得超过20万
        for (Object object: objects) {
            IndexRequest request = new IndexRequest(index);
            request.source(JSON.toJSONString(object), XContentType.JSON);
            bulkRequest.add(request);
        }
        try {
            response=restHighLevelClientGroupMsg.bulk(bulkRequest,RequestOptions.DEFAULT);
        } catch (IOException e) {
            log.error(e.getMessage(),e);
        }
        return response.hasFailures();
    }

    /**
     * 根据经纬度查询范围查找location 经纬度字段，distance 距离中心范围KM，lat  lon 圆心经纬度
     * @param index
     * @param longitude
     * @param latitude
     * @param distance
     * @return
     */
    public  SearchResponse geoDistanceQuery(String index, Float  longitude, Float latitude, String distance) throws IOException {

        if(longitude == null || latitude == null){
            return null;
        }
        //拼接条件
        BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();
//        QueryBuilder isdeleteBuilder = QueryBuilders.termQuery("isdelete", false);
        // 以某点为中心，搜索指定范围
        GeoDistanceQueryBuilder distanceQueryBuilder = new GeoDistanceQueryBuilder("location");
        distanceQueryBuilder.point(latitude, longitude);
        //查询单位：km
        distanceQueryBuilder.distance(distance, DistanceUnit.KILOMETERS);
        boolQueryBuilder.filter(distanceQueryBuilder);
//        boolQueryBuilder.must(isdeleteBuilder);

        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        searchSourceBuilder.query(boolQueryBuilder);

        SearchRequest searchRequest = new SearchRequest(index);
        searchRequest.source(searchSourceBuilder);

        SearchResponse searchResponse = restHighLevelClientGroupMsg.search(searchRequest,RequestOptions.DEFAULT);
        return searchResponse;
    }

    /**
     * 获取低水平客户端
     * @return
     */
    public  RestClient getLowLevelClient() {
        return restHighLevelClientGroupMsg.getLowLevelClient();
    }


    /**
     * 高亮结果集 特殊处理
     * map转对象 JSONObject.parseObject(Tools.getStrEmpty(JSONObject.toJSONString(map)), Content.class)
     * @param searchResponse
     * @param highlightField
     */
    private  ESResult setSearchResponse(SearchResponse searchResponse, String highlightField) {
        //解析结果，包括总数totalHits和数据datalist
        ESResult esResult = new ESResult();
        //解析数据LIST
        ArrayList<Map<String,Object>> dataList = new ArrayList<>();

        //添加总记录数
        long totalHits = searchResponse.getHits().getTotalHits().value;
        esResult.setTotalHits(totalHits);

        for (SearchHit hit : searchResponse.getHits().getHits()) {
            Map<String, HighlightField> high = hit.getHighlightFields();
            HighlightField title = high.get(highlightField);
            Map<String, Object> sourceAsMap = hit.getSourceAsMap();//原来的结果
            //解析高亮字段,将原来的字段换为高亮字段
            if (title!=null){
                Text[] texts = title.fragments();
                String nTitle="";
                for (Text text : texts) {
                    nTitle+=text;
                }
                //替换
                sourceAsMap.put(highlightField,nTitle);
            }
            dataList.add(sourceAsMap);
        }
        esResult.setDataList(dataList);
        return esResult;
    }

    /**
     * 高亮结果集 特殊处理
     * map转对象 JSONObject.parseObject(Tools.getStrEmpty(JSONObject.toJSONString(map)), Content.class)
     * @param searchResponse
     * @param highlightField
     * @param highlightField
     */
    private  ESResult setSearchResponseByDistinct(SearchResponse searchResponse, String highlightField, String distinctFieldCount) {
        //解析结果，包括总数totalHits和数据datalist
        ESResult esResult = new ESResult();
        //解析数据LIST
        ArrayList<Map<String,Object>> dataList = new ArrayList<>();

        //添加总记录数
        long totalHits = searchResponse.getHits().getTotalHits().value;
        Aggregations aggregations = searchResponse.getAggregations();
        if(aggregations != null ){
            ParsedCardinality fileCount = aggregations.get(distinctFieldCount);
            if(fileCount != null){
                totalHits = NumberUtil.parseLong(fileCount.getValueAsString());
            }
        }

        esResult.setTotalHits(totalHits);

        for (SearchHit hit : searchResponse.getHits().getHits()) {
            Map<String, HighlightField> high = hit.getHighlightFields();
            HighlightField title = high.get(highlightField);
            Map<String, Object> sourceAsMap = hit.getSourceAsMap();//原来的结果
            //解析高亮字段,将原来的字段换为高亮字段
            if (title!=null){
                Text[] texts = title.fragments();
                String nTitle="";
                for (Text text : texts) {
                    nTitle+=text;
                }
                //替换
                sourceAsMap.put(highlightField,nTitle);
            }
            dataList.add(sourceAsMap);
        }
        esResult.setDataList(dataList);
        return esResult;
    }


    /**
     * 查询并分页
     * @param index          索引名称
     * @param query          查询条件
     * @param size           文档大小限制
     * @param fields         需要显示的字段，逗号分隔（缺省为全部字段）
     * @param sortField      排序字段
     * @param highlightField 高亮字段
     * @return Map<String, Object>: 包括总数totalHits和数据dataList
     */
    public ESResult searchListData(String index,
                                                     SearchSourceBuilder query,
                                                     Integer size,
                                                     Integer from,
                                                     String fields,
                                                     String sortField,
                                                     String highlightField, Long merId){
        SearchRequest request = new SearchRequest(index);

        //解决ES相同条件搜索多次返回结果不一样
        if(merId != null){
            request.preference(String.valueOf(merId));
        }

        //打印条件
        //log.info(query.query().toString());

        SearchSourceBuilder builder = query;
        if (StringUtils.isNotEmpty(fields)){
            //只查询特定字段。如果需要查询所有字段则不设置该项。
            builder.fetchSource(new FetchSourceContext(true,fields.split(","),Strings.EMPTY_ARRAY));
        }
        //分页已经设置，不需要在设置
        //from = from <= 0 ? 0 : from*size;
        //设置确定结果要从哪个索引开始搜索的from选项，默认为0

        if (from.intValue() + size.intValue() > ESConstant.TOTAL_MAX) {
            throw new BusinessException(ResponseCode.BUSINESS_EXCEPTION, String.format("只能查看前[%s]条数据, 建议缩小查询范围", ESConstant.TOTAL_MAX));
        }

        if(from != null){
            builder.from(from);
        }

        if(size != null){
            builder.size(size);
        }

        if (StringUtils.isNotEmpty(sortField)){
            //排序字段，注意如果proposal_no是text类型会默认带有keyword性质，需要拼接.keyword
            builder.sort(sortField+".keyword", SortOrder.ASC);
        }

        //默认查询不能超于1万条，如果数据超过1万条，则设置如下参数
        builder.trackTotalHits(true);

        if(StringUtils.isNotEmpty(highlightField)){
            //高亮
            HighlightBuilder highlight = new HighlightBuilder();
            highlight.field(highlightField);
            //关闭多个高亮
            highlight.requireFieldMatch(false);
            highlight.preTags("<span style='color:red'>");
            highlight.postTags("</span>");
            builder.highlighter(highlight);
        }

        //不返回源数据。只有条数之类的数据。
        //builder.fetchSource(false);
        request.source(builder);

        try {
            SearchResponse response = restHighLevelClientGroupMsg.search(request, RequestOptions.DEFAULT);
            if (response.status().getStatus() == 200) {
                // 解析对象
                return setSearchResponse(response, highlightField);
            }
        } catch (IOException e) {
            log.error(e.getMessage(),e);
        }

        return null;
    }

    /**
     * 去重查询并分页
     * @param index          索引名称
     * @param query          查询条件
     * @param size           文档大小限制
     * @param fields         需要显示的字段，逗号分隔（缺省为全部字段）
     * @param sortField      排序字段
     * @param highlightField 高亮字段
     * @param distinctFieldCount 去重之后得总数字段
     * @return Map<String, Object>: 包括总数totalHits和数据dataList
     */
    public ESResult distinctSearchListData(String index,
                                   SearchSourceBuilder query,
                                   Integer size,
                                   Integer from,
                                   String fields,
                                   String sortField,
                                   String highlightField, Long merId, String distinctFieldCount){
        SearchRequest request = new SearchRequest(index);

        //解决ES相同条件搜索多次返回结果不一样
        if(merId != null){
            request.preference(String.valueOf(merId));
        }

        //打印条件
        //log.info(query.query().toString());

        SearchSourceBuilder builder = query;
        if (StringUtils.isNotEmpty(fields)){
            //只查询特定字段。如果需要查询所有字段则不设置该项。
            builder.fetchSource(new FetchSourceContext(true,fields.split(","),Strings.EMPTY_ARRAY));
        }
        //分页已经设置，不需要在设置
        //from = from <= 0 ? 0 : from*size;
        //设置确定结果要从哪个索引开始搜索的from选项，默认为0

        if (from.intValue() + size.intValue() > ESConstant.TOTAL_MAX) {
            throw new BusinessException(ResponseCode.BUSINESS_EXCEPTION, String.format("只能查看前[%s]条数据, 建议缩小查询范围", ESConstant.TOTAL_MAX));
        }

        if(from != null){
            builder.from(from);
        }

        if(size != null){
            builder.size(size);
        }

        if (StringUtils.isNotEmpty(sortField)){
            //排序字段，注意如果proposal_no是text类型会默认带有keyword性质，需要拼接.keyword
            builder.sort(sortField+".keyword", SortOrder.ASC);
        }

        //默认查询不能超于1万条，如果数据超过1万条，则设置如下参数
        builder.trackTotalHits(true);

        if(StringUtils.isNotEmpty(highlightField)){
            //高亮
            HighlightBuilder highlight = new HighlightBuilder();
            highlight.field(highlightField);
            //关闭多个高亮
            highlight.requireFieldMatch(false);
            highlight.preTags("<span style='color:red'>");
            highlight.postTags("</span>");
            builder.highlighter(highlight);
        }

        //不返回源数据。只有条数之类的数据。
        //builder.fetchSource(false);
        request.source(builder);

        try {
            SearchResponse response = restHighLevelClientGroupMsg.search(request, RequestOptions.DEFAULT);
            if (response.status().getStatus() == 200) {
                // 解析对象
                return setSearchResponseByDistinct(response, highlightField, distinctFieldCount);
            }
        } catch (IOException e) {
            log.error(e.getMessage(),e);
        }

        return null;
    }

    /**
     * 查询数据总数
     *
     * @param index
     * @param query
     * @return
     * @throws IOException
     */
    public long searchTotal(String index, SearchSourceBuilder query){

        //默认查询不能超于1万条，如果数据超过1万条，则设置如下参数
        query.trackTotalHits(true);

        //返回总记录数
        long totalHits = 0;
        SearchRequest searchRequest = new SearchRequest(index);
        searchRequest.source(query.size(10)); //一次查询10条

        try {
            SearchResponse searchResponse = restHighLevelClientGroupMsg.search(searchRequest, RequestOptions.DEFAULT);
            //总记录数
            if(searchResponse != null){
                SearchHits searchHits = searchResponse.getHits();
                if(searchHits != null){
                    TotalHits totalHits_object = searchHits.getTotalHits();
                    if(totalHits_object != null){
                        totalHits = totalHits_object.value;
                    }
                }
            }
        } catch (Exception e) {
            log.error("searchTotal field failed!", e.getMessage());
        }

        return totalHits;
    }



    /**
     * 使⽤游标滚动查询全部数据，返回SearchHit集合
     *
     * @param index
     * @param query
     * @return
     * @throws IOException
     */
    public List<Map<String,Object>> scrollSearchAll(String index, SearchSourceBuilder query) {

        //解析数据LIST
        List<Map<String,Object>> dataList = new ArrayList<>();

        try {
            //默认查询不能超于1万条，如果数据超过1万条，则设置如下参数
            query.trackTotalHits(true);

            SearchRequest searchRequest = new SearchRequest(index);

            //解决ES相同条件搜索多次返回结果不一样
            if(query != null){
                searchRequest.preference(query.hashCode()+"");
            }

            searchRequest.source(query.size(500)); //一次查询500条
            Scroll scroll = new Scroll(new TimeValue(600000));
            searchRequest.scroll(scroll);

            SearchResponse searchResponse = restHighLevelClientGroupMsg.search(searchRequest, RequestOptions.DEFAULT);
            String scrollId = searchResponse.getScrollId();
            SearchHit[] hits = searchResponse.getHits().getHits();

            while (ArrayUtils.isNotEmpty(hits)) {
                for (SearchHit hit : hits) {
                    Map<String, Object> sourceAsMap = hit.getSourceAsMap();//原来的结果
                    dataList.add(sourceAsMap);
                }

                // 再次发送请求,并使用上次搜索结果的ScrollId
                SearchScrollRequest searchScrollRequest = new SearchScrollRequest(scrollId);
                searchScrollRequest.scroll(scroll);
                SearchResponse searchScrollResponse = restHighLevelClientGroupMsg.searchScroll(searchScrollRequest, RequestOptions.DEFAULT);

                scrollId = searchScrollResponse.getScrollId();
                hits = searchScrollResponse.getHits().getHits();
            }

            //及时清除快照，释放资源
            ClearScrollRequest clearScrollRequest = new ClearScrollRequest();
            clearScrollRequest.addScrollId(scrollId);
            restHighLevelClientGroupMsg.clearScroll(clearScrollRequest, RequestOptions.DEFAULT);
        } catch (IOException e) {
            log.error(e.getMessage(),e);
        }
        return dataList;
    }


    /**
     * 使⽤游标滚动查询全部数据，返回SearchHit集合
     *
     * @param index
     * @param query
     * @return
     * @throws IOException
     */
    public List<Map<String,Object>> scrollSearchAll(String index, SearchSourceBuilder query, Long merId) {

        //解析数据LIST
        List<Map<String,Object>> dataList = new ArrayList<>();

        try {
            //默认查询不能超于1万条，如果数据超过1万条，则设置如下参数
            query.trackTotalHits(true);

            SearchRequest searchRequest = new SearchRequest(index);

            //解决ES相同条件搜索多次返回结果不一样
            if(merId != null){
                searchRequest.preference(String.valueOf(merId));
            }


            searchRequest.source(query.size(500)); //一次查询500条
            Scroll scroll = new Scroll(new TimeValue(600000));
            searchRequest.scroll(scroll);

            SearchResponse searchResponse = restHighLevelClientGroupMsg.search(searchRequest, RequestOptions.DEFAULT);
            String scrollId = searchResponse.getScrollId();
            SearchHit[] hits = searchResponse.getHits().getHits();

            while (ArrayUtils.isNotEmpty(hits)) {
                for (SearchHit hit : hits) {
                    Map<String, Object> sourceAsMap = hit.getSourceAsMap();//原来的结果
                    dataList.add(sourceAsMap);
                }

                // 再次发送请求,并使用上次搜索结果的ScrollId
                SearchScrollRequest searchScrollRequest = new SearchScrollRequest(scrollId);
                searchScrollRequest.scroll(scroll);
                SearchResponse searchScrollResponse = restHighLevelClientGroupMsg.searchScroll(searchScrollRequest, RequestOptions.DEFAULT);

                scrollId = searchScrollResponse.getScrollId();
                hits = searchScrollResponse.getHits().getHits();
            }

            //及时清除快照，释放资源
            ClearScrollRequest clearScrollRequest = new ClearScrollRequest();
            clearScrollRequest.addScrollId(scrollId);
            restHighLevelClientGroupMsg.clearScroll(clearScrollRequest, RequestOptions.DEFAULT);
        } catch (IOException e) {
            log.error(e.getMessage(),e);
        }
        return dataList;
    }

    /**
     * 查询数据总数-根据某个字段去重
     *
     * @param esIndex
     * @param countField
     * @return
     * @throws IOException
     */
    public long countDistinctField(String esIndex, String countField, SearchSourceBuilder sourceBuilder) {
        long count = 0;
        if (StringUtils.isBlank(esIndex) || StringUtils.isBlank(countField)) {
              return count;
          }

        SearchRequest searchRequest = new SearchRequest();
        searchRequest.indices(esIndex);
        //去重别名
        String identifier = countField + "_" + UUID.randomUUID().toString();
        AggregationBuilder aggregationBuilder = AggregationBuilders.cardinality(identifier).field(countField);
        sourceBuilder.aggregation(aggregationBuilder);
        sourceBuilder.size(0);
        searchRequest.source(sourceBuilder);
        try {
            // 查询
            SearchResponse response = restHighLevelClientGroupMsg.search(searchRequest, RequestOptions.DEFAULT);
            Aggregations aggregations = response.getAggregations();
            if(aggregations != null){
                Cardinality  cardinality_sex_result = aggregations.get(identifier);
                if(cardinality_sex_result != null){
                    long total_value = cardinality_sex_result.getValue();
                    return total_value;
                }
            }
          } catch (Exception e) {
              log.error("Count field failed!", e.getMessage());
          }
        return 0;
    }

}
